package com.xhg.mqtt.core;

import com.xhg.mqtt.constant.MqttConstants;
import com.xhg.mqtt.constant.MsgConstants;
import com.xhg.mqtt.constant.ResultCodeConstants;
import com.xhg.mqtt.dto.EventCodeEnum;
import com.xhg.mqtt.dto.MqttConfigInfoDTO;
import com.xhg.mqtt.dto.MqttMessagePb;
import com.xhg.mqtt.listener.MessageManager;
import com.xhg.mqtt.listener.MqttLoggerFactory;
import com.xhg.mqtt.util.CommonUtil;
import com.xhg.mqtt.util.PropertiesUtil;
import com.xhg.mqtt.util.ProtoBufUtil;
import com.xhg.mqtt.util.SSLSocketManager;
import java.util.HashMap;
import java.util.Map;
import javax.net.ssl.SSLSocketFactory;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * mqtt连接
 *
 * @author E.T
 * @version 1.0
 * @Description
 * @date 2018-11-07 10:38
 * @updateby
 * @updatedate
 * @since 1.0
 */
public class MqttClient {

    private static org.eclipse.paho.client.mqttv3.MqttClient client;
    private MqttConfigInfoDTO mqttInfo;
    private String[] topics;
    private String deviceId;
    private String version;
    private static Map<String, String> map = new HashMap<>();
    private MqttClient() {
    }
    private static class MqttClientHolder {
        private final static MqttClient instance = new MqttClient();
    }
    public static MqttClient getInstance() {
        return MqttClientHolder.instance;
    }
    public static org.eclipse.paho.client.mqttv3.MqttClient getMqttInstance() {
        return client;
    }

    /**
     * 获取连接实例
     *
     * @param mqttInfo mqtt信息
     * @param deviceId 设备id
     * @param version 版本号
     */
    public void connection(MqttConfigInfoDTO mqttInfo, String deviceId, String version) {
        if (mqttInfo == null) {
            MqttLoggerFactory.error(MsgConstants.MQTTINFO_IS_NULL);
        }
        map.put("{province}", mqttInfo.getProvince());
        map.put("{city}", mqttInfo.getCity());
        map.put("{area}", mqttInfo.getArea());
        map.put("{clientId}", deviceId);
        map.put("{eventCode}", EventCodeEnum.DEVICE_WILL.getCode());
        String[] topics = CommonUtil.replaceTopics(mqttInfo.getSubscribeTopics(), map);
        this.topics = topics;
        this.mqttInfo = mqttInfo;
        this.deviceId = deviceId;
        this.version = version;
        connect();
    }

    /**
     * 连接mqtt
     */
    private void connect()  {
        String[] urls = mqttInfo.getUris();
        if (urls == null || urls.length <= 0) {
            urls = PropertiesUtil.MQTT_HOST.split(",");
        }
        try {
            client = new org.eclipse.paho.client.mqttv3.MqttClient(urls[0], deviceId,
                new MemoryPersistence());
            MqttConnectOptions options = getMqttConnectOptions();
            client.setCallback(new Callback());
            client.setTimeToWait(PropertiesUtil.MQTT_TIME_TO_WAIT);
            client.connect(options);
            subTopic();
        } catch (MqttException e) {
            MessageManager.connectComplete(false);
            MqttLoggerFactory.error(MsgConstants.CONN_EXCEPTION, e);
        } catch (Exception e) {
            MessageManager.connectComplete(false);
            MqttLoggerFactory.error(MsgConstants.SYS_EXCEPTION, e);
        }
    }

    /**
     * 设置连接mqtt参数
     * @return
     * @throws Exception
     */
    private MqttConnectOptions getMqttConnectOptions() throws Exception {
        String[] urls = mqttInfo.getUris();
        int authType = mqttInfo.getAuthType();
        String username = mqttInfo.getUsername();
        String pwd = mqttInfo.getPassword();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(urls);
        if (MqttConstants.AUTH_CRTPA == authType) {
            SSLSocketFactory factory = SSLSocketManager.getSSLSocktet(pwd);
            options.setSocketFactory(factory);
        }
        if (MqttConstants.AUTH_USERPWD == authType) {
            options.setUserName(username);
            if (null != pwd && !"".equals(pwd)) {
                options.setPassword(pwd.toCharArray());
            }
        }
        options.setCleanSession(PropertiesUtil.CLEAN_SESSION);
        options.setConnectionTimeout(PropertiesUtil.MQTT_TIMEOUT);
        options.setKeepAliveInterval(PropertiesUtil.MQTT_KEEP_ALIVE);
        options.setMaxInflight(PropertiesUtil.MAXINFLIGHT);
        byte[] bytes = builderWill(deviceId, version);
        //设置遗言（断开后自动发送消息）
        options.setWill(CommonUtil.replaceTopic(mqttInfo.getPushTopic(), map), bytes,
            MqttConstants.QOS,
            MqttConstants.RETAINED_FALSE);
        //设置自动重连
        options.setAutomaticReconnect(true);
        return options;
    }

    /**
     * 生成遗嘱
     */
    private byte[] builderWill(String clientId, String version) throws Exception {
        MqttMessagePb.MqttHead.Builder head = ProtoBufUtil
            .builderHead(EventCodeEnum.DEVICE_WILL.getCode(), CommonUtil.getUUID(""),
                clientId,
                version, 0,
                ResultCodeConstants.SUCCESS, MsgConstants.OFFLINE_EXCEPTION);
        return ProtoBufUtil.builderMessage(head, null);
    }


    /**
     * 订阅消息
     */
    public void subTopic() {
        //订阅默认消息 qos =1
        int[] qos = new int[topics.length];
        for (int i = 0; i < qos.length; i++) {
            qos[i] = MqttConstants.QOS;
        }
        try {
            client.subscribe(topics, qos);
        } catch (MqttException e) {
            MqttLoggerFactory.error(MsgConstants.SUB_ERROR, e);
        }
    }
}
